package com.sg.java.apps;

import com.sg.java.PropertiesUtil;
import com.sg.java.ResourcePath;
import org.apache.commons.csv.CSVFormat;
import org.apache.commons.csv.CSVPrinter;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import redis.clients.jedis.Jedis;
import redis.clients.jedis.Pipeline;

import java.io.FileOutputStream;
import java.io.OutputStreamWriter;
import java.nio.charset.StandardCharsets;
import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.util.Properties;

/**
 * 数据中台台账数据入redis做电压数据计算时候要用的关联关系
 */
public class YhToPb {

    private static final Logger log = LoggerFactory.getLogger(YhToPb.class);


    public static void main(String[] args) throws Exception {
        log.info("从数据中台dws读取用户关联关系数据存入redis");
        Properties dwsDataProp = PropertiesUtil.createAndLoadPropertiesFromFileOrResource(null, ResourcePath.dws_data_jdbc_properties);
        log.info("数据中台dws配置：" + dwsDataProp.toString());
        Properties redisProp = PropertiesUtil.createAndLoadPropertiesFromFileOrResource(null, ResourcePath.redis_properties);
        log.info("redis配置：" + redisProp.toString());
        try (
                Jedis jedis = new Jedis(redisProp.getProperty("host"), Integer.parseInt(redisProp.getProperty("port")), 1000 * 60);
                Connection conn = DriverManager.getConnection(
                        dwsDataProp.getProperty("url"),
                        dwsDataProp.getProperty("user"),
                        dwsDataProp.getProperty("password")
                )
        ) {
            jedis.auth(redisProp.getProperty("password"));
            //开通管道
            Pipeline jp = jedis.pipelined();
//            redisSetAc();
            jedis.select(0);


            String filePath = "/tmp/yhTopb.csv";
            CSVFormat formator = CSVFormat.DEFAULT
                    .withRecordSeparator("\n")
                    .withDelimiter(',');
            FileOutputStream fos = new FileOutputStream(filePath);
            OutputStreamWriter fileWriter = new OutputStreamWriter(fos, StandardCharsets.UTF_8);
            CSVPrinter printer = new CSVPrinter(fileWriter, formator);


            PreparedStatement ps;
            int batchSize = 500000;
            int index = 0;
            boolean hasMore = true;
            while (hasMore) {
                int count = 0;
                String sql = getPbsql(batchSize, index);
                ps = conn.prepareStatement(sql);
                log.info(ps.toString());
                ResultSet rs = ps.executeQuery();
                while (rs.next()) {
                    String key = "YHTOPB@" + rs.getString(1);
                    String value = rs.getString(2);
//                    jedis.set(key, value);

//                    jp.set(key, value);
                    printer.printRecord(rs.getString(1) + "," + value);
                    count++;
                }
                //管道异步执行
//                jp.sync();
                index += batchSize;
                hasMore = count == batchSize;
            }

            log.info("全部同步结束");
            jp.close();
            printer.flush();
            printer.close();
//            jedis.close();
        }
    }


    private static String getPbsql(int batchSize, int index) {


        StringBuilder stringBuilder = new StringBuilder();


        stringBuilder.append(" select rela.meter_id,g.tg_id  from (select  a.*,bb.sfnw  from dwd_cms.un14_02_cms_g_tran a, (select aa.*  from ( select  a.obj_id,a.ywdw,a.dydj,a.yxzt,a.tyrq,a.sbmc,a.sbbm,a.xh,a.edrl,a.sfnw,a.ds " +
                " from  dwd_pms.un11_02_pms_t_sb_zwyc_zsbyq a  where (a.zcxz in ('03', '04')  and  a.zyfl in('3') and  a.yxzt in('20')) " +
                " union all " +
                " select  a.obj_id,a.ywdw ,a.dydj,a.yxzt ,a.tyrq,a.sbmc,a.sbbm,a.xh,a.edrl,a.sfnw,a.ds from  dwd_pms.un11_02_pms_t_sb_znyc_pdbyq a  where  (a.zcxz in ('03', '04') and a.zyfl in('3') and  a.yxzt in('20'))) aa)  bb  where  a.pms_equip_id  = bb.obj_id )  ccc , un_cms.un14_02_cms_c_mp_cqzj g,un_cms.un14_02_cms_c_meter_mp_rela_cqzj rela  where ccc.tg_id = g.tg_id  and g.usage_type_code ='01' and ccc.run_status_code = '01'  and ccc.pub_priv_flag = '01'  and g.status_code  in('01','02') and g.type_code = '01' " +
                " and rela.mp_id = g.mp_id ");
        stringBuilder.append(" limit " + batchSize + " offset " + index);


        return stringBuilder.toString();


    }


}
